package com.ww.springboot.boot.rxjava.rxjava2;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import rx.Observable;
import rx.functions.Action1;
import rx.functions.Func2;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
public class CreaterOperatorDemo {
    /**
     * 演示just的基本使用
     */
    @Test
    public void demo() throws InterruptedException {

        //jush();
        //
        //from();
        //
        //range();
        //
        //interval();
        //
        //defer();
        //simpleWindowObserver();
        //windowObserver();
        //timeWindowObserverDemo();
        //
        scan();
    }

    public void scan() throws InterruptedException {
//        Observable observable = Observable.just(1, 2, 3, 4, 5);
//
//        //按顺序 处理相邻的两个数 并返回处理结果
//        observable.scan((Func2<Integer, Integer, Integer>) (o, o2) -> o + o2)
//                //跳过指定数量请求
//                .skip(3)
//                .subscribe(o -> System.out.println("scan-- " + o));

        Observable
                //间隔
                .interval(100, TimeUnit.MILLISECONDS)
                //造数据
                .map(i -> i)
                .scan((o, o1) -> o + o1)
                //跳过指定数量请求
                //.skip(3)
                .subscribe(o -> System.out.println("scan-- " + o));

        Thread.sleep(Integer.MAX_VALUE);
    }

    public void simpleWindowObserver() {
        List<Integer> srcList = Arrays.asList(10, 11, 20, 21, 30, 31);
        Observable.from(srcList)
                .window(3) //以固定数量分组
                .flatMap(o -> o.toList())
                .subscribe(list -> log.info(list.toString()));
    }

    public void windowObserver() {

        List<Integer> srcList = Arrays.asList(10, 11, 20, 21, 30, 31);
        Observable.from(srcList)
                .window(3, 1)
                .flatMap(o -> o.toList())
                .subscribe(list -> log.info(list.toString()));
    }

    public void timeWindowObserverDemo() throws InterruptedException {

        Observable eventStream = Observable
                .interval(100, TimeUnit.MILLISECONDS);
        eventStream.window(300, TimeUnit.MILLISECONDS)
                .flatMap(o -> ((Observable<Integer>) o).toList())
                .subscribe(list -> log.info(list.toString()));
        Thread.sleep(Integer.MAX_VALUE);
    }


    public void range() {
        //逐一发一组范围内的整数序列
        Observable.range(1, 10)
                .subscribe(i -> log.info("just int->" + i));
    }

    public void interval() throws InterruptedException {
        Observable
                .interval(100, TimeUnit.MILLISECONDS)
                .subscribe(aLong -> log.info(aLong.toString()));

        Thread.sleep(Integer.MAX_VALUE);
    }

    /**
     * 订阅值在被订阅时才弹射(被确定)
     */
    public void defer() {
        AtomicInteger foo = new AtomicInteger(100);
        Observable observable = Observable.just(foo.get());
        /**
         *延迟创建
         */
        Observable dObservable = Observable.defer(

                () -> Observable.just(foo.get()));

        /**
         *修改对象的值
         */
        foo.set(200);
        /**
         *有观察者订阅
         */
        observable.subscribe(
                integer -> log.info("just emit {}", String.valueOf(integer)));
        /**
         *有观察者订阅
         */
        dObservable.subscribe(
                integer -> log.info("defer just emit {}", String.valueOf(integer)));

    }

    private void from() {
        //逐一发送一个数组中的每一个元素
        String[] items = {"a", "b", "c", "d", "e", "f"};
        Observable.from(items)
                .subscribe(s -> log.info("just string->" + s));

        //逐一发送迭代器中的每一个元素
        Integer[] array = {1, 2, 3, 4};
        List<Integer> list = Arrays.asList(array);
        Observable.from(list)
                .subscribe(i -> log.info("just int->" + i));
    }

    private void jush() {
        //发送一个字符串"hello world"
        Observable.just("hello world")
                .subscribe(s -> log.info("just string->" + s));
        //逐一发送1,2,3,4四个整数
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
                .subscribe(i -> log.info("just int->" + i));
    }

}